package week13.topic05;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;

public class PulsarTest {
    public static void main(String[] args) throws Exception {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://192.168.75.131:6650")
                .build();
        //生产者
        Producer<byte[]> producer = client.newProducer()
                .topic("my-topic")
                .create();

        // 然后你就可以发送消息到指定的broker 和topic上：
        producer.send("My message".getBytes());

        //消费者
        Consumer consumer = client.newConsumer()
                .topic("my-topic")
                .subscriptionName("my-subscription-test")
                .subscribe();

        while (true){
            Message msg = consumer.receive();
            try {
                // Do something with the message
                System.out.println("Message received: " + new String(msg.getData()));
                // Acknowledge the message so that it can be deleted by the message broker
                consumer.acknowledge(msg);
                Thread.sleep(100);
            } catch (Exception e) {
                // Message failed to process, redeliver later
                consumer.negativeAcknowledge(msg);
            }

        }
    }
}
